package sc.analysis.ppo;



import java.io.IOException;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import sc.writable.PriceData;



/**
 * Mapper of the Percentage Price Oscillator (PPO) technical analysis. It's like
 * the MAC-D but it has relative values instead of absolute values.
 * 
 * @author Quentin Rossé
 *
 */
public class PpoMapper extends Mapper<Text, PriceData, Text, PpoData>{
	
	/**
	 * number of days for the shorter EMA
	 */
	private static final int PERIOD_SHORT_EMA = 12;
	
	/**
	 * number of days for the longer EMA
	 */
	private static final int PERIOD_LONG_EMA = 26;
	
	/**
	 * number of days for the EMA of the PPO values
	 */
	private static final int PERIOD_PPO = 9;
	
	/*
	 * multipliers for the EMA
	 */
	private static final float MULT_SHORT_EMA = 2f / (PERIOD_SHORT_EMA+1f);
	private static final float MULT_LONG_EMA = 2f / (PERIOD_LONG_EMA+1f);
	private static final float MULT_PPO_EMA = 2f / (PERIOD_PPO+1f);
	
	/**
	 * output data
	 */
	private static PpoData ppod = new PpoData();
	
	
	public void map(Text key, PriceData value, Context context) throws IOException, InterruptedException{
		float[] closePrices = value.getClosePrices().toFloatArray();
		int arraySize = closePrices.length;
		
		System.out.println();
		System.out.println();
		System.out.println(key);
		
		if(arraySize >= 3 * PERIOD_LONG_EMA){
			float[] emaShort =  ema(closePrices, MULT_SHORT_EMA, PERIOD_SHORT_EMA, 2*PERIOD_SHORT_EMA);
			float[] emaLong =  ema(closePrices, MULT_LONG_EMA, PERIOD_LONG_EMA, 2*PERIOD_LONG_EMA);
			float[] ppo = new float[emaShort.length];
			
			
			
			
			for(int i=1; i<=emaShort.length; i++){
				ppo[emaShort.length-i] = (emaShort[emaShort.length-i] -
				                          emaLong[emaLong.length-i]) /
				                          emaLong[emaLong.length-i] * 100f;
			}
			
			for(int i=0; i<ppo.length; i++)
				System.out.println("ppo["+i+"] : " + ppo[i]);
			
			float[] signal = ema(ppo, MULT_PPO_EMA, PERIOD_PPO, ppo.length - PERIOD_PPO);
				
			ppod.setAll(ppo[ppo.length-1], 
					    ppo[ppo.length-2], 
					    ppo[ppo.length-1] - signal[signal.length -1],
					    ppo[ppo.length-2] - signal[signal.length -2]);
			context.write(key, ppod);
		}
		
	}
	
	private float[] ema(float[] prices, 
			            float mutiplier, 
			            int period,
			            int nbDays){
		
		System.out.println("######period : " + period + "   // size : " + prices.length);
		
		float[] ema = new float[nbDays];
		
		for(int i=prices.length - nbDays - period; 
		        i < prices.length - nbDays; i++){
			ema[0] += prices[i];
			System.out.println("i : " + i + "  price : " + prices[i]);
		}
		ema[0] /= period;
		
		for(int i=1; i < nbDays; i++){
			ema[i] = mutiplier * prices[prices.length - nbDays + i] +
			         (1f - mutiplier) * ema[i-1];
			System.out.println("i : " + (prices.length - nbDays + i) + 
					"  price : " + prices[prices.length - nbDays + i] +
					"  ema : " + ema[i]);
		}
		return ema;
	}
	
	@Override
	public void setup(Context context){}
	
	@Override
	public void cleanup(Context context){}
	
}
